<?php
/**
 * Created by PhpStorm.
 * User: sxg
 * Date: 2023/7/3
 * Time: 11:16
 * Doc: https://mcxiaoke.gitbooks.io/mqtt-cn/content/mqtt/01-Introduction.html
 */

namespace Socket\Ms\Protocols;

class Mqtt implements Protocols {

    public $control_packet_type = 0;
    public $control_packet_type_name = '';

    public static $control_packets = [
        0b00010000 => "CONNECT_PACKET_TYPE",
        0b00100000 => "CONNACK_PACKET_TYPE",
        0b00110000 => "PUBLISH_PACKET_TYPE",
        0b01000000 => "PUBACK_PACKET_TYPE",
        0b01010000 => "PUBREC_PACKET_TYPE",
        0b01100000 => "PUBREL_PACKET_TYPE",
        0b01110000 => "PUBCOMP_PACKET_TYPE",
        0b10000000 => "SUBSCRIBE_PACKET_TYPE",
        0b10010000 => "SUBACK_PACKET_TYPE",
        0b10100000 => "UNSUBSCRIBE_PACKET_TYPE",
        0b10110000 => "UNSUBACK_PACKET_TYPE",
        0b11000000 => "PINGREQ_PACKET_TYPE",
        0b11010000 => "PINGRESP_PACKET_TYPE",
        0b11100000 => "DISCONNECT_PACKET_TYPE",
    ];

    public $remain_length = 0;
    public $header_len = 0;
    public $fix_header_len = 0;

    public $protocolName = '';
    public $protocolLevel = 0;

    public $cleanSessionFlag = 0;
    public $willFlag = 0;
    public $willQosFlag = 0;
    public $willRetainFlag = 0;
    public $passwordFlag = 0;
    public $userNameFlag = 0;
    public $keepAlive = 0;

    public $clientIdentifier = 0;
    public $willTopic = '';
    public $willMessage = '';
    public $userName = '';
    public $password = '';

    public $unsubscribeReserve = 0;

    //订阅报文标识符
    public $subscribePacketIdentifier = 0;
    //取消订阅报文标识符
    public $unsubscribePacketIdentifier = 0;
    //订阅主题
    public $subscribeTopics = [];

    //重发标志
    public $publishPackDup = 0;
    //服务质量等级
    public $publishPackQos = 0;
    //保留位
    public $publishPackRetain = 0;
    //主题名
    public $publishPackTopicName = '';
    //主题名长度
    public $publishPackTopicNameLen = 0;
    //publish报文标识符
    public $publishPacketIdentifierMSB = 0;
    public $publishPacketIdentifierLSB = 0;
    //publish消息内容
    public $publishPacketMessage = '';
    //publish消息长度
    public $publishPacketMessageLen = 0;
    //PUBREL报文标识符
    public $pubrelPacketIdentifierMSB = 0;
    public $pubrelPacketIdentifierLSB = 0;
    //PUBREC报文标识符
    public $pubrecPacketIdentifierMSB = 0;
    public $pubrecPacketIdentifierLSB = 0;

    const CONNECT_PACK_TYPE     =   0b00010000;
    const CONNACK_PACK_TYPE     =   0b00100000;
    const PUBLISH_PACK_TYPE     =   0b00110000;
    const PUBACK_PACK_TYPE      =   0b01000000;
    const PUBREC_PACK_TYPE      =   0b01010000;
    const PUBREL_PACK_TYPE      =   0b01100000;
    const PUBCOMP_PACK_TYPE     =   0b01110000;
    const SUBSCRIBE_PACK_TYPE   =   0b10000000;
    const SUBACK_PACK_TYPE      =   0b10010000;
    const UNSUBSCRIBE_PACK_TYPE =   0b10100000;
    const UNSUBACK_PACK_TYPE    =   0b10110000;
    const PINGREQ_PACK_TYPE     =   0b11000000;
    const PINGRESP_PACK_TYPE    =   0b11010000;
    const DISCONNECT_PACK_TYPE  =   0b11100000;

    public function Len($data) {
        // TODO: Implement Len() method.
        if (strlen($data) < 2) {
            return false;
        }

        $this->header_len = 2;
        $this->fix_header_len = 2;

        $this->control_packet_type = ord($data[0]) & 0xf0;
        $this->remain_length = ord($data[1]);
        $this->control_packet_type_name = self::$control_packets[$this->control_packet_type];
        $ret = false;

        switch ($this->control_packet_type) {
            case self::CONNECT_PACK_TYPE:
                $ret = $this->decodeConnectPackLen($data);
                break;
            case self::CONNACK_PACK_TYPE:
                break;
            case self::PUBLISH_PACK_TYPE:
                $ret = $this->decodePublishPackLen($data);
                break;
            case self::PUBACK_PACK_TYPE:
                $ret = $this->decodePubAckPacketLen();
                break;
            case self::PUBREC_PACK_TYPE:
                $ret = $this->decodePubRecPacketLen($data);
                break;
            case self::PUBREL_PACK_TYPE:
                $ret = $this->decodePubRelPacketLen($data);
                break;
            case self::PUBCOMP_PACK_TYPE:
                $this->remain_length = 2;   //这里原来写的是0 导致客户端在服务质量等级为2的条件下给订阅者返回PUBCOMP时出错
                // 接收缓冲区数据没有清除 接收到客户端的PUBREC时没有正常解析数据包
                $this->header_len += 2;
                return true;
            case self::SUBSCRIBE_PACK_TYPE:
                //订阅主题
                $ret = $this->decodeSubscribePackLen($data);
                break;
            case self::UNSUBSCRIBE_PACK_TYPE:
                $ret = $this->decodeUnsubscribePackLen($data);
                break;
            case self::PINGREQ_PACK_TYPE:
                //服务端必须发送 PINGRESP报文响应客户端的PINGREQ报文
                $this->remain_length = 0;
                $this->header_len += 2;
                $ret = true;
                break;
            case self::DISCONNECT_PACK_TYPE:
                $this->remain_length = 0;
                $this->header_len = 2;
                $ret = true;
                break;
            default:
                $ret = true;
                break;
        }
        return $ret;
    }

    public function calculateRemainLen($data) {
        $remainLen = $firstRemainLenByte= $this->remain_length;
        if ($remainLen >= 128) {
            $this->header_len += 1;
            $secondRemainLenByte = ord($data[2]);
            $remainLen = $secondRemainLenByte * 128 + ($firstRemainLenByte & 0x7f);
        }
        if ($remainLen >= 16384) {
            $this->header_len += 1;
            $thirdRemainLenByte = ord($data[3]);
            $remainLen = ($thirdRemainLenByte * 128 + (ord($data[2]) & 0x7f)) * 128 + ($firstRemainLenByte & 0x7f);
        }
        if ($remainLen >= 209712) {
            $this->header_len += 1;
            $fourthRemainLenByte = ord($data[4]);
            $remainLen = (($fourthRemainLenByte * 128 + (ord($data[3]) & 0x7f)) * 128 + (ord($data[2]) & 0x7f)) * 128 + ($firstRemainLenByte & 0x7f);
        }
        $this->remain_length = $remainLen;
        $this->fix_header_len = $this->header_len;
    }

    public function decodeConnectPackLen($data) : bool {
        $this->calculateRemainLen($data);
        if (strlen($data) < $this->header_len + $this->remain_length) {
            return false;
        }

        $currentPos = 0;
        $data = substr($data, $this->header_len);
        $unpack = unpack("nLen", $data);
        $protocolNameLen = $unpack['Len'];
        $currentPos += 2;
        $this->protocolName = substr($data, $currentPos, $protocolNameLen);
        $currentPos += $protocolNameLen;
        if (strncmp($this->protocolName, "MQTT", 4) != 0) {
            //协议名不对
            return true;
        }

        $this->protocolLevel = ord($data[$currentPos]);
        if ($this->protocolLevel != 4) {
            //协议及别不正确
            return true;
        }
        $currentPos += 1;

        $connectFlags = ord($data[$currentPos]);
        $this->cleanSessionFlag = ($connectFlags & 0b00000010) == 0b00000010 ? 1 : 0;
        $this->willFlag = ($connectFlags & 0b00000100) == 0b00000100 ? 1 : 0;
        $this->willRetainFlag = ($connectFlags & 0b00100000) == 0b00100000 ? 1 : 0;
        $this->passwordFlag = ($connectFlags & 0b01000000) == 0b01000000 ? 1 : 0;
        $this->userNameFlag = ($connectFlags & 0b10000000) == 0b10000000 ? 1 : 0;
        $this->willQosFlag = ($connectFlags & 0b00011000) >> 3;
        $currentPos += 1;

        $keepAliveMSB = ord($data[$currentPos]);
        $currentPos += 1;
        $keepAliveLSB = ord($data[$currentPos]);
        $currentPos += 1;
        $this->keepAlive |= $keepAliveLSB << 0;
        $this->keepAlive |= $keepAliveMSB << 8;
        $this->header_len += $currentPos;
        return true;
    }

    public function decodeSubscribePackLen($data) : bool {
        $this->calculateRemainLen($data);
        if (strlen($data) < $this->header_len + $this->remain_length) {
            return false;
        }

        $data = substr($data, $this->header_len);
        //解析报文标识符
        $packetIdentifierMSB = ord($data[0]);
        $packetIdentifierLSB = ord($data[1]);
        $this->subscribePacketIdentifier |= $packetIdentifierMSB << 8;
        $this->subscribePacketIdentifier |= $packetIdentifierLSB << 0;
        $this->header_len += 2;
        return true;
    }

    public function decodeUnsubscribePackLen($data) : bool {
        $this->calculateRemainLen($data);
        if (strlen($data) < $this->header_len + $this->remain_length) {
            return false;
        }
        //UNSUBSCRIBE报文固定报头的第3,2,1,0位是保留位且必须分别设置为0,0,1,0
        $this->unsubscribeReserve = ord($data[0]) & 0x0f;
        if ($this->unsubscribeReserve != 0x02) {
            return true;
        }
        //可变报头
        $data = substr($data, $this->header_len);
        $this->unsubscribePacketIdentifier = unpack('nL', $data)['L'];
        $this->header_len += 2;
        return true;
    }

    public function decodePublishPackLen($data) : bool {
        $this->calculateRemainLen($data);
        if (strlen($data) < $this->header_len + $this->remain_length) {
            return false;
        }
        //如果DUP标志被设置为0，表示这是客户端或服务端第一次请求发送这个PUBLISH报文。
        //客户端或服务端请求重发一个PUBLISH报文时，必须将DUP标志设置为1
        //如果DUP标志被设置为1，表示这可能是一个早前报文请求的重发。
        $firstByte = ord($data[0]);
        $this->publishPackDup = (($firstByte & 0b00001000) == 0b00001000) ? 1 : 0;
        $this->publishPackQos = ($firstByte & 0b00000110) >> 1;
        //客户端发给服务端的PUBLISH报文的保留标志被设置为1，服务端必须存储这个应用消息和它的服务质量等级，以便它可以被分发给未来的主题名匹配的订阅者
        //【这个就懒得搞了啊 不存储了】
        $this->publishPackRetain = ($firstByte & 0x01) == 1 ? 1 : 0;
        $data = substr($data, $this->header_len);

        //解析主题名相关
        $this->publishPackTopicNameLen = unpack("nL", $data)['L'];
        $this->publishPackTopicName = substr($data, 2, $this->publishPackTopicNameLen);
        $this->header_len += (2 + $this->publishPackTopicNameLen);
        //报文标识符
        //只有当QoS等级是1或2时，报文标识符（Packet Identifier）字段才能出现在PUBLISH报文中
        if ($this->publishPackQos > 0) {
            $data = substr($data, 2 + $this->publishPackTopicNameLen);
            $this->publishPacketIdentifierMSB = ord($data[0]);
            $this->publishPacketIdentifierLSB = ord($data[1]);
            $this->header_len += 2;
        }
        return true;
    }

    public function decodePubAckPacketLen() : bool {
        $this->remain_length = 2;
        $this->header_len += 2;
        return true;
    }

    public function decodePubRelPacketLen($data) : bool {
        $this->remain_length = 2;
        if (strlen($data) < $this->header_len + $this->remain_length) {
            return false;
        }
        $data = substr($data, $this->header_len);
        $this->pubrelPacketIdentifierMSB = ord($data[0]);
        $this->pubrelPacketIdentifierLSB = ord($data[1]);
        $this->header_len += 2;
        return true;
    }

    public function decodePubRecPacketLen($data) : bool {
        $this->remain_length = 2;
        if (strlen($data) < $this->header_len + $this->remain_length) {
            return false;
        }
        $data = substr($data, $this->header_len);
        $this->pubrecPacketIdentifierMSB = ord($data[0]);
        $this->pubrecPacketIdentifierLSB = ord($data[1]);
        $this->header_len += 2;
        return true;
    }

    public function encode($data = '') {
        // TODO: Implement encode() method.
        $frame = '';
        switch ($this->control_packet_type) {
            case self::CONNECT_PACK_TYPE:
                if ($this->protocolLevel != 4) {
                    //如果发现不支持的协议级别，服务端必须给发送一个返回码为0x01（不支持的协议级别）的CONNACK报文响应
                    $frame = $this->responseConnAckPack(0x01);
                } else {
                    $frame = $this->responseConnAckPack(0x00);
                }
                break;
            case self::PINGREQ_PACK_TYPE:
                $frame = $this->responsePINGRESP();
                break;
            case self::SUBSCRIBE_PACK_TYPE:
                $frame = $this->responseSUBACK();
                break;
            case self::UNSUBSCRIBE_PACK_TYPE:
                $frame = $this->responseUNSUBACK();
                break;
            case self::PUBLISH_PACK_TYPE:
                if ($this->publishPackQos == 1) {
                    $frame = $this->responsePUBACK();
                } else if ($this->publishPackQos == 2) {
                    $frame = $this->responsePUBREC();
                }
                break;
            case self::PUBREL_PACK_TYPE:
                $frame = $this->responsePUBCOMP();
                break;
            case self::PUBREC_PACK_TYPE:
                $frame = $this->responsePUBREL();
                break;
        }
        return [strlen($frame), $frame];
    }

    public function decode($data = '') {
        // TODO: Implement decode() method.
        switch ($this->control_packet_type) {
            case self::CONNECT_PACK_TYPE:
                $this->decodeConnectPacket($data);
                return '';
            case self::PUBREL_PACK_TYPE:
            case self::PUBREC_PACK_TYPE:
            case self::PUBACK_PACK_TYPE:
            case self::PINGREQ_PACK_TYPE:
            case self::PUBCOMP_PACK_TYPE:
            case self::DISCONNECT_PACK_TYPE:
                return '';
            case self::SUBSCRIBE_PACK_TYPE:
                $this->decodeSubscribePacket($data);
                return '';
            case self::UNSUBSCRIBE_PACK_TYPE:
                $this->decodeUnsubscribePacket($data);
                return '';
            case self::PUBLISH_PACK_TYPE:
                //有效载荷包含将被发布的应用消息。数据的内容和格式是应用特定的。
                //有效载荷的长度这样计算：用固定报头中的剩余长度字段的值减去可变报头的长度。
                //包含零长度有效载荷的PUBLISH报文是合法的。
                $this->decodePublishPacket($data);
                return $this->publishPacketMessage;
        }
    }

    public function msgLen($data = '') {
        // TODO: Implement msgLen() method.
        return $this->fix_header_len + $this->remain_length;
    }

    public function decodeConnectPacket($data) {
        $data = substr($data, $this->header_len);
        $currentPos = 0;
        $clientIdentifierLen = unpack("nLen", $data)['Len'];
        $this->clientIdentifier = substr($data, 2, $clientIdentifierLen);
        $currentPos += 2 + $clientIdentifierLen;

        //解析遗嘱主题
        $data = substr($data, $currentPos);
        $currentPos = 0;
        if ($this->willFlag) {
            $willTopicLen = unpack("nLen", $data)['Len'];
            $this->willTopic = substr($data, 2, $willTopicLen);
            $currentPos += 2 + $willTopicLen;
            $data = substr($data, $currentPos);
            $currentPos = 0;
        }
        if ($this->willQosFlag) {
            $willMessageLen = unpack("nLen", $data)['Len'];
            $this->willMessage = substr($data, 2, $willMessageLen);
            $currentPos += 2 + $willMessageLen;
            $data = substr($data, $currentPos);
            $currentPos = 0;
        }
        if ($this->userNameFlag) {
            $userNameLen = unpack("nLen", $data)['Len'];
            $this->userName = substr($data, 2, $userNameLen);
            $currentPos += 2 + $userNameLen;
            $data = substr($data, $currentPos);
        }
        if ($this->passwordFlag) {
            $passwordLen = unpack("nLen", $data)['Len'];
            $this->password = substr($data, 2, $passwordLen);
        }
    }

    public function decodeSubscribePacket($data) {
        $data = substr($data, $this->header_len);
        //循环解析
        while (strlen($data)) {
            $topicFilterLen = unpack("nL", $data)['L'];
            $topic = substr($data, 2, $topicFilterLen);
            $qos = ord($data[2 + $topicFilterLen]);
            $data = substr($data, 2 + $topicFilterLen + 1);
            $this->subscribeTopics[$topic] = $qos;
        }
    }


    public function decodeUnsubscribePacket($data) {
        $data = substr($data, $this->header_len);
        //循环解析
        while (strlen($data)) {
            $topicFilterLen = unpack("nL", $data)['L'];
            $topic = substr($data, 2, $topicFilterLen);
            if (isset($this->subscribeTopics[$topic])) {
                unset($this->subscribeTopics[$topic]);
            }
            $data = substr($data, 2 + $topicFilterLen);
        }
    }

    public function decodePublishPacket($data) {
        $this->publishPacketMessage = substr($data, $this->header_len);
        $this->publishPacketMessageLen = strlen($this->publishPacketMessage);
    }

    public function responseConnAckPack($returnCode) : string {
        $frame = chr(self::CONNACK_PACK_TYPE).chr(0x02);
        //如果服务端收到清理会话（CleanSession）标志为1的连接
        //除了将CONNACK报文中的返回码设置为0之外，还必须将CONNACK报文中的当前会话设置（Session Present）标志为0
        if ($this->cleanSessionFlag == 1) {
            $frame .= chr(0x00);
        } else {
            //如果服务端收到一个CleanSession为0的连接，当前会话标志的值取决于服务端是否已经保存了ClientId对应客户端的会话状态
            //1.如果服务端已经保存了会话状态，它必须将CONNACK报文中的当前会话标志设置为1
            //2.如果服务端没有已保存的会话状态，它必须将CONNACK报文中的当前会话设置为0
            $frame .= chr(0x01);
        }
        $frame .= chr($returnCode);
        return $frame;
    }

    public function responsePINGRESP() : string {
        return chr(self::PINGRESP_PACK_TYPE).chr(0x00);
    }

    public function responseSUBACK() : string {
        $remainLen = 2 + count($this->subscribeTopics);
        $payload = '';
        foreach ($this->subscribeTopics ?? [] as $topic => $qos) {
            $payload .= chr($qos);
        }
        //报文标识符 MSB
        $MSB = $this->subscribePacketIdentifier >> 8 & 0xff;
        //报文标识符 LSB
        $LSB = $this->subscribePacketIdentifier >> 0 & 0xff;
        return chr(self::SUBACK_PACK_TYPE).chr($remainLen).chr($MSB).chr($LSB).$payload;
    }

    public function responseUNSUBACK() : string {
        $payload = '';
        foreach ($this->subscribeTopics ?? [] as $topic => $qos) {
            $payload .= chr($qos);
        }
        //取消订阅报文标识符 MSB
        $MSB = $this->unsubscribePacketIdentifier >> 8 & 0xff;
        //取消订阅报文标识符 LSB
        $LSB = $this->unsubscribePacketIdentifier >> 0 & 0xff;
        //UNSUBACK报文没有有效载荷。
        return chr(self::UNSUBACK_PACK_TYPE).chr(0x02).chr($MSB).chr($LSB);
    }

    public function responsePUBLISHPACK($Qos) : string {
        $dup = 0x00;
        $qos = 0x00;
        $retain = 0x00;
        $realQos = min($Qos, $this->publishPackQos);
        if ($this->publishPackDup == 1) {
            $dup = 0x08;
        }
        if ($realQos == 1) {
            $qos = 0b00000010;
        } else if ($realQos == 2) {
            $qos = 0b00000100;
        }
        if ($this->publishPackRetain == 1) {
            $retain = 0x01;
        }
        $frame = chr(self::PUBLISH_PACK_TYPE|$dup|$qos|$retain);
        $remainLength = 2 + $this->publishPackTopicNameLen + $this->publishPacketMessageLen;
        print_r("remainLength:".$remainLength."\r\n");
        //如果服务质量大于0 还有两个字节的报文标识符
        if ($realQos > 0) {
            $remainLength += 2;
        }
        //对剩余长度进行编码
        $this->calculateEncodePublishRemainLen($remainLength, $frame);
        //可变报头按顺序包含主题名和报文标识符。
        $frame .= pack("n", $this->publishPackTopicNameLen);
        $frame .= $this->publishPackTopicName;
        if ($realQos > 0) {
            //Qos质量等级大于0有报文标识符
            $frame .= chr($this->publishPacketIdentifierMSB).chr($this->publishPacketIdentifierLSB);
        }
        //有效载荷
        $frame .= $this->publishPacketMessage;
        return $frame;
    }

    public function responsePUBACK() : string {
        return chr(self::PUBACK_PACK_TYPE).chr(0x02).chr($this->publishPacketIdentifierMSB).chr($this->publishPacketIdentifierLSB);
    }

    public function responsePUBREC() : string {
        return chr(self::PUBREC_PACK_TYPE).chr(0x02).chr($this->publishPacketIdentifierMSB).chr($this->publishPacketIdentifierLSB);
    }

    public function responsePUBREL() : string {
        return chr(self::PUBREL_PACK_TYPE|0x02).chr(0x02).chr($this->pubrecPacketIdentifierMSB).chr($this->pubrecPacketIdentifierLSB);
    }

    public function responsePUBCOMP() : string {
        return chr(self::PUBCOMP_PACK_TYPE).chr(0x02).chr($this->pubrelPacketIdentifierMSB).chr($this->pubrelPacketIdentifierLSB);
    }


    public function calculateEncodePublishRemainLen($remainLength, &$frame) {
        if ($remainLength <= 127) {
            $frame .= pack("C", $remainLength);
        } else if ($remainLength <= 16383) {
            //两个字节存储剩余长度
            $firstByte = 0x80 + ($remainLength % 128);
            $secondByte = floor($remainLength / 128);
            $frame .= pack("CC", $firstByte, $secondByte);
        } else if ($remainLength <= 2097151) {
            //三个字节存储剩余长度
            $firstByte = 0x80 + ($remainLength % 128);
            $secondByte = 0x80 + (($remainLength / 128) % 128);
            $thirdByte = floor($remainLength / 128) / 128;
            $frame .= pack("CCC", $firstByte, $secondByte, $thirdByte);
        } else {
            //四个字节存储剩余长度
            $firstByte = 0x80 + ($remainLength % 128);
            $secondByte = 0x80 + (($remainLength / 128) % 128);
            $thirdByte = 0x80 + (($remainLength / 128 / 128) % 128);
            $fourthByte = ($remainLength / 128 / 128) / 128;
            $frame .= pack("CCCC", $firstByte, $secondByte, $thirdByte, $fourthByte);
        }
    }
}